package com.leo.biz.service.msgcenter;

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

/**
 * Created by leo on 2016/11/16.
 *功能：本partitioner是均匀分发到各个partition中
 *
 */
public class MsgPartitioner implements Partitioner {

    public MsgPartitioner(VerifiableProperties props) {
    }

    public static  int lastPartitonId = 0;



    @Override
    /*注:只有KeyedMessage中指定有key时，才会进行partitioner函数
    *  KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, key,msg);
        producer.send(data);
    * */
    public int partition(Object key, int numPartitions) {
        lastPartitonId=(lastPartitonId+1)%numPartitions;
        return lastPartitonId;
    }

    public int partition_default(Object key, int numPartitions) {
        int partition = 0;
        if (key instanceof String) {
            String strkey=(String)key;
            int offset = strkey.lastIndexOf('.');
            if (offset > 0) {
                partition = Integer.parseInt(strkey.substring(offset + 1)) % numPartitions;
            }
        }else{
           // partition = obj.toString().length() % numPartitions;
            partition =(int)Math.round((Math.random() * 1000000)) % numPartitions;
            System.out.println(partition);
        }
        return partition;
    }
}
